package com.gugee.ins.data.collector.media.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.media.InsMediaDao;
import com.gugee.ins.common.dal.service.product.InsProductDao;
import com.gugee.ins.common.dal.service.product.InsRelMediaProductDao;
import com.gugee.ins.common.dal.spider.InsMediaSpiderClient;
import com.gugee.ins.common.model.mysql.blogger.InsBlogger;

import com.gugee.ins.common.model.mysql.product.InsProduct;
import com.gugee.ins.common.model.mysql.product.InsRelMediaProduct;

import com.gugee.ins.common.model.spider.app.media.*;

import com.gugee.ins.data.collector.media.config.InsMediaConfig;
import com.gugee.ins.data.collector.media.service.InsMediaService;
import com.gugee.ins.data.collector.media.service.InsProductService;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.ins.data.common.msg.InsProductMessage;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.ForkJoinPool;

/**
 * insBlogger服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsProductServiceImpl implements InsProductService {

    /**
     * 配置
     */
    @Autowired
    InsMediaConfig insMediaConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * 爬虫接口
     */
    @Autowired
    InsMediaSpiderClient insMediaSpiderClient;

    /**
     *
     */
    @Autowired
    InsMediaService insMediaService;

    /**
     * InsBloggerExtDao
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     * insProductDao
     */
    @Autowired
    InsProductDao insProductDao;

    /**
     * insRelMediaProductDao
     */
    @Autowired
    InsRelMediaProductDao insRelMediaProductDao;

    /**
     * InsMediaDao
     */
    @Autowired
    InsMediaDao insMediaDao;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsBlogger;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsProductMessage> kafkaTemplateInsProductMsg;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    private String workName = "insAppProductMediaCollect-worker";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(insMediaConfig.getInsAppProductMediaCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 标记商品帖子的
     * @param productIds
     */
    @Override
    public void handleInsAppProductMediaCollect(List<Long> productIds) {
        try {
            forkJoinPool.submit(() -> productIds.parallelStream().forEach(productId -> {
                try {
                    //采集播主关注人数据
                    runSpiderProductInsMedia(productId);
                } catch (Exception ex) {
                    log.error("Ins app product Media collect error. productId:{}, error :{}" ,
                            productId,ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("Ins app product Media collect  thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 标记商品帖子的
     * @param insBloggerMessages
     */
    @Override
    public void handleInsAppProductMediaCl(List<InsBloggerMessage> insBloggerMessages) {
        try {
            forkJoinPool.submit(() -> insBloggerMessages.parallelStream().forEach(msg -> {
                try {
                    //采集播主关注人数据
                    runSpiderProductInsMedia(msg.getUid());
                } catch (Exception ex) {
                    log.error("Ins app product Media collect error. productId:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("Ins app product Media collect  thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 采集更新帖子数据
     * @param productId
     */
    public void runSpiderProductInsMedia(long productId){
        long start = System.currentTimeMillis();
        // 根据uid爬取信息
        String pageInfo;

        // 30天时间区间
        long timeRang = start / 1000 - insMediaConfig.getLatestDays() * 24 * 3600;
        InsProduct insProduct = insProductDao.findByProductId(productId);
        if(insProduct == null){
            return;
        }
        long lastCollectTime = start / 1000 - insMediaConfig.getInsProductMediaIntervalHour() * 3600;
        if(insProduct.getLastMediaTagCollectTime() < lastCollectTime){
            List<InsRelMediaProduct> relMp = new ArrayList<>(10);

            pageInfo = spiderProductMedia(productId,"",timeRang,relMp,insProduct);
            while(!StringUtils.isEmpty(pageInfo)){
                pageInfo = spiderProductMedia(productId, pageInfo,timeRang,relMp,insProduct);
            }

            // 更新最近帖子提及时间,以及采集时间
            insProductDao.updateMediaTagTime(insProduct,true);

            // 保存关系
            if(!relMp.isEmpty()){
                saveRelMp(relMp);
            }
        }
        if ((System.currentTimeMillis() - start) / 1000 > insMediaConfig.getMediaDebugTimeLimit()) {
            log.info("Ins app product Media collect productId:{} use time:{}s", productId,(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 循环拉帖子
     * @param productId
     * @param endCursor
     * @param timeRang
     * @return
     */
    public String spiderProductMedia(long productId,String endCursor,long timeRang,List<InsRelMediaProduct> relMp,InsProduct insProduct){
        //根据uid爬取信息
        List<InsAppMediaBody> edges;
        InsAppGetProductMediaResponse insAppGetProductMediaResponse;
        String pageInfo;

        try {
            insAppGetProductMediaResponse = insMediaSpiderClient.getProductMarkedMediaList(productId,endCursor);
            edges = insAppGetProductMediaResponse.getItems();
            if(edges != null && !edges.isEmpty()){
                long uid = 0L;
                String userName = "";
                String picUrl = "";
                boolean isPrivate = false;
                long followerCount = 0L;
                for(InsAppMediaBody edge : edges){
                    if(edge != null){
                        if(edge.getTakenTime() > timeRang){
                            try{
                                if (edge.getUser() != null && edge.getUser().getUid() != null) {
                                    uid = edge.getUser().getUid();
                                    userName = edge.getUser().getUsername();
                                    picUrl = edge.getUser().getPicUrl();
                                    isPrivate = edge.getUser().getIsPrivate() != null && edge.getUser().getIsPrivate();
                                }
                                InsBlogger insBlogger = insBloggerDao.findByUid(uid);
                                if(insBlogger != null){
                                    followerCount = insBlogger.getFollowerCount();
                                }else if(uid > 0){
                                    // 发送优先收入提及商品的帖子作者信息
                                    sendPriorityCollectBloggerMsg(uid,userName,picUrl,isPrivate, InsBloggerMsgTypeCodeEnum.CLP_SRC_MEDIA_TAG_PRODUCT.getCode());
                                }

                                //保存帖子及相关信息
                                insMediaService.saveMedia(edge,followerCount);

                            }catch (Exception e){
                                log.error("ins app product media merge error,msg:{}",e.getMessage());
                            }
                            // 添加 帖子和商品的关系
                            getProductRel(uid,edge.getShortcode(),edge.getTakenTime(),productId,relMp);
                            if(edge.getTakenTime() > insProduct.getLastMediaTagTime()){
                                insProduct.setLastMediaTagTime(edge.getTakenTime());
                            }
                        }else{
                            //直接返回空，不再爬取
                            return null;
                        }
                    }
                }
            }
            pageInfo = insAppGetProductMediaResponse.getNextMaxId();
        }catch (Exception e){
            pageInfo = null;
            log.error("spider ins app product media collect error, msg :{}",e.getMessage());
        }
        return pageInfo;
    }

    /**
     * 优先采集播主信息
     * @param uid
     * @param userName
     * @param picUrl
     * @param isPrivate
     */
    public void sendPriorityCollectBloggerMsg(long uid,String userName,String picUrl,boolean isPrivate,int srcTypeCode){
        if(uid > 0 && insBloggerDao.checkCollecting(uid,false)){
            //这里需要添加验证
            InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
            insBloggerMessage.setUid(uid);
            insBloggerMessage.setUserName(userName);
            insBloggerMessage.setPicUrl(StringUtils.isEmpty(picUrl) ? "" : picUrl);
            insBloggerMessage.setRanking(isPrivate);
            insBloggerMessage.setRetryLimitTimes(insMediaConfig.getInsBloggerPriorityRetryTimes());
            insBloggerMessage.setRetryTimes(insMediaConfig.getInsBloggerPriorityRetryTimes());
            insBloggerMessage.setTypeCode(srcTypeCode);

            kafkaTemplateInsBlogger.send(insMediaConfig.getInsBloggerPriorityCollectOnlineTopic(),insBloggerMessage);
        }
    }

    /**
     * 获取商品相关数据
     * @param uid
     * @param shortcode
     * @param mediaTakeTime
     * @param productId
     * @param relMp
     */
    public void getProductRel(long uid,String shortcode, long mediaTakeTime,long productId,List<InsRelMediaProduct> relMp){
        // 帖子商品关联信息
        InsRelMediaProduct insRelMediaProduct = new InsRelMediaProduct();
        insRelMediaProduct.setShortcode(shortcode);
        insRelMediaProduct.setTakenTime(mediaTakeTime);
        insRelMediaProduct.setCreateTime(System.currentTimeMillis() / 1000);
        insRelMediaProduct.setProductId(productId);
        insRelMediaProduct.setUid(uid);
        insRelMediaProduct.setIsSelf(false);
        relMp.add(insRelMediaProduct);
    }

    /**
     * 保存商品和帖子关系
     * @param relMp
     */
    @Override
    public void saveRelMp(List<InsRelMediaProduct> relMp){
        for(InsRelMediaProduct insRelMediaProduct : relMp){
            try{
                Long rmpId = insRelMediaProductDao.findExist(insRelMediaProduct);
                if(rmpId == null){
                    // 保存商品和帖子关系
                    insRelMediaProductDao.save(insRelMediaProduct);
                    // 更新 帖子的带货状态
                    insMediaDao.updateProductTagType(insRelMediaProduct.getShortcode());
                } else{
                    insRelMediaProduct.setId(rmpId);
                    insRelMediaProductDao.updateById(insRelMediaProduct);
                }
            }catch (Exception e){
                log.error("save rel Media Product error,productId :{}, shortcode:{},msg:{}",
                        insRelMediaProduct.getProductId(),insRelMediaProduct.getShortcode(),e.getMessage());
            }
        }
    }

    /**
     * 发送分析商品分类消息
     */
    @Override
    public void sendAnalysisProductCategoryMsg(List<InsProduct> products){
        if(products != null && !products.isEmpty()){
            InsProductMessage insProductMessage = new InsProductMessage();
            products.forEach(product -> {
                insProductMessage.setDescription(product.getDescription());
                insProductMessage.setProductName(product.getProductName());
                insProductMessage.setExternalUrl(product.getExternalUrl());
                kafkaTemplateInsProductMsg.send(insMediaConfig.getInsAnalysisProductCategoryOnlineTopic(),insProductMessage);
            });
        }
    }
}
